package com.bw.jointest;

import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.co.ProcessJoinFunction;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.Collector;
import scala.Tuple2;

public class FlinkDataStreamJoinTest {
    public static void main(String[] args) throws Exception {

        //1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        //2.读取两个端口数据创建流并提取时间戳生成WaterMark
        SingleOutputStreamOperator<Bean1> stream1 = env.socketTextStream("hadoop101", 8888)
                .map(line -> {
                    String[] fields = line.split(",");
                    return new Bean1(fields[0], fields[1], Long.parseLong(fields[2]));
                }).assignTimestampsAndWatermarks(WatermarkStrategy.<Bean1>forMonotonousTimestamps()
                        .withTimestampAssigner(new SerializableTimestampAssigner<Bean1>() {
                            @Override
                            public long extractTimestamp(Bean1 element, long recordTimestamp) {
                                return element.getTs() * 1000L;
                            }
                        }));
        SingleOutputStreamOperator<Bean2> stream2 = env.socketTextStream("hadoop101", 9999)
                .map(line -> {
                    String[] fields = line.split(",");
                    return new Bean2(fields[0], fields[1], Long.parseLong(fields[2]));
                }).assignTimestampsAndWatermarks(WatermarkStrategy.<Bean2>forMonotonousTimestamps()
                        .withTimestampAssigner(new SerializableTimestampAssigner<Bean2>() {
                            @Override
                            public long extractTimestamp(Bean2 element, long recordTimestamp) {
                                return element.getTs() * 1000L;
                            }
                        }));

        //3.双流JOIN
        SingleOutputStreamOperator<Tuple2<Bean1, Bean2>> joinDS = stream1.keyBy(Bean1::getId)
                .intervalJoin(stream2.keyBy(Bean2::getId))
                .between(Time.seconds(-5), Time.seconds(5))
                //下面两个不加 左闭右闭
                //加lowerBoundExclusive 左开右闭
                //加upperBoundExclusive 左闭右开
                //两个都加就是开区间
//                .lowerBoundExclusive()
//                .upperBoundExclusive()
                .process(new ProcessJoinFunction<Bean1, Bean2, Tuple2<Bean1, Bean2>>() {
                    @Override
                    public void processElement(Bean1 left, Bean2 right, Context ctx, Collector<Tuple2<Bean1, Bean2>> out) throws Exception {
                        out.collect(new Tuple2<>(left, right));
                    }
                });

        //4.打印
        joinDS.print();

        //5.启动任务
        env.execute();

        /*
            1.linux开启；
                nc -lk 8888
                    1001,zhangsan,6



                nc -lk 9999
                    1001,male,1

             2.运行程序FlinkDataStreamJoinTest

             8888 回车
             9999 回车


                第二次 9999 发送数据看控制台效果：
                    1001,female,2

                第二次 8888 发送数据看控制台效果不会打印因为id 1002：
                    1002,lisi,10

                第二三次 9999 发送数据看控制台效果有数据打印，为什么? ：
                    1001,male,3

         */

    }
}
